package com.lg.distributed.transaction.message.config;

import com.lg.distributed.transaction.message.coordinator.CompleteCorrelationData;
import com.lg.distributed.transaction.message.coordinator.DBCoordinator;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

/**
 * 自定义 ，设置可靠消息的回调
 *
 * @author by Mr. Li 2020/12/11 22:59
 */
@Configuration
@Slf4j
@EnableRabbit
public class RabbitTemplateConfig {

    @Autowired
    private ApplicationContext applicationContext;

    @Autowired
    private RabbitTemplate rabbitTemplate;
    //private RabbitTemplate rabbitTemplate;
    //
    //public RabbitTemplateConfig(RabbitTemplate rabbitTemplate) {
    //    this.rabbitTemplate = rabbitTemplate;
    //}

    private boolean returnFlag = false;

    @PostConstruct
    public void initRabbitTemplate() {
        // 1. 设置服务收到消息 回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {

            /**
             * 服务器收到消息就回调
             *
             * @param correlationData 当前消息的唯一关联数据（包含消息唯一id)
             * @param ack 消息是否成功收到
             * @param cause 失败的原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {

                if (returnFlag) {
                    log.error("mq发送错误，无对应的的交换机,confirm回掉,ack={},correlationData={} cause={} returnFlag={}",
                            ack, correlationData, cause, returnFlag);
                }

                log.info("The server successfully received the message，ack ： {}", ack);
                if (ack && correlationData != null) {
                    // 成功收到信息，并且已经持久化，将状态写入到磁盘
                    String msgId = correlationData.getId();
                    // 这里需要判断，持久化方案是什么
                    // 这里获取前面设置的 持久化类型。
                    String dbCoordinator = ((CompleteCorrelationData) correlationData).getDbCoordinator();
                    DBCoordinator coordinator = (DBCoordinator) applicationContext.getBean(dbCoordinator);
                    // 设置消息状态--成功
                    coordinator.setMsgSuccess(msgId);
                } else {
                    log.info("Message failed ,routingKey : {},cause: {}", correlationData.getId(), cause);
                }
            }
        });

        // 2. 消息没有到达指定的队列
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            /**
             *
             *消息没有到达指定的队列触发，这个回调如果被触发会比confirm回调先触发
             *@param returnedMessage message    投递失败的消息信息
             *                         replyCode  回复码
             *                         replyText  回复的文本内容
             *                         exchange   当时消息发给哪个交换机
             *                         routingKey 当时消息发给哪个交换机时的路由键
             *           在spring boot 2.4 上述内容被整合成了 returnedMessage
             */
            @Override
            public void returnedMessage(ReturnedMessage returnedMessage) {
                String messageId = returnedMessage.getMessage().getMessageProperties().getMessageId();
                int replyCode = returnedMessage.getReplyCode();
                String replyText = returnedMessage.getReplyText();
                String exchange = returnedMessage.getExchange();
                String routingKey = returnedMessage.getRoutingKey();

                log.error("return回调，没有找到任何匹配的队列！message id:{},replyCode{},replyText:{},"
                        + "exchange:{},routingKey{}", messageId, replyCode, replyText, exchange, routingKey);
                returnFlag = true;
            }
        });
    }

}
